package epoch.game.core.distr;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import epoch.game.core.actor.Actor;
import epoch.game.core.context.Context;
import epoch.game.core.distr.call.Call;
import epoch.game.core.distr.call.CallReturn;
import epoch.game.core.distr.config.NodeConfig;
import epoch.game.core.distr.netty.NodeAcceptor;
import epoch.game.core.distr.netty.NodeConnector;
import epoch.game.core.distr.service.IService;
import epoch.game.core.distr.service.ServiceLocator;
import epoch.game.core.exception.ConfigException;
import epoch.game.core.exception.SysException;
import epoch.game.core.message.IMessage;
import epoch.game.core.timer.GameTimer;
import epoch.game.core.util.BeanUtil;

/**
 * 服务节点，负责服务器之间的通讯
 * 
 * @author will
 * @date 2015年11月18日 上午9:51:39
 */
public class Node {

	public static Logger logger = LogManager.getLogger(Node.class);
	
	//所有的远程节点
	private Map<Integer, RemoteNode> remoteNodes = new ConcurrentHashMap<Integer, RemoteNode>();
	// 节点线程,负责监测与其他节点的连接状态检测等工作
	private Actor nodeActor;
	
	// 节点中挂接的线程
	public Map<Integer, ThreadContext> threadContexts = new ConcurrentHashMap<Integer, ThreadContext>();
	// 所挂接的服务
	private static Map<Integer, IService> services = new ConcurrentHashMap<Integer, IService>();
	
	// id
	private int id;
	// 名称
	private String name;
	// 绑定的ip地址
	private String host;
	// 端口
	private int port;
	// 监听接受器
	private INodeAcceptor acceptor;
	// 连接器
	INodeConnector connector;
	// 消息处理器
	private INodeHandler handler;
	// 消息通道
//	private INodeChannel channel;
	// 是否可用，服务器启动OK之后才设置该值，表示可以接受节点间的通讯
	private volatile boolean available = false;
	// 线程id生成器
	private AtomicInteger threadContextIdGen;
	// 节点类型
	NodeType type;
	
	public Node(int id, String name, String host, int port, NodeType type) {
		this.id = id;
		this.name = name;
		this.host = host;
		this.port = port;
		this.type = type;
		this.connector = new NodeConnector();
		nodeActor = Context.createActor(new Actor(), "NodeMain");
		threadContextIdGen = new AtomicInteger(1);
		//注册定时器更新时间，节点tick
	}
	
	public Node(NodeConfig config) {
		this.id = config.getId();
		this.name = config.getName();
		this.host = config.getHost();
		this.port = config.getPort();
		this.type = config.getType();
		this.connector = new NodeConnector();
		nodeActor = Context.createActor(new Actor(), "NodeMain");
		threadContextIdGen = new AtomicInteger(1);
		//注册定时器更新时间，节点tick
	}

	public int getServerId() {
		return NodeConfig.serverId(id);
	}
	
	public int getId() {
		return id;
	}
	
	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getHost() {
		return host;
	}
	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}

	public INodeHandler getHandler() {
		return handler;
	}

	public void setHandler(INodeHandler handler) {
		this.handler = handler;
	}

	/**
	 * 是否可用，服务器启动OK之后才设置该值，表示可以接受节点间的通讯
	 * @return
	 */
	public boolean isAvailable() {
		return available;
	}
	
	/**
	 * 设置节点是否可用，服务器启动OK之后才设置该值，表示可以接受节点间的通讯
	 * @param available
	 */
	public void setAvailable(boolean available) {
		this.available = available;
	}

	private void connectRemotes() {
		List<RemoteNode> disconnectNodes = new ArrayList<>();
		Iterator<Entry<Integer, RemoteNode>> iterator = remoteNodes.entrySet().iterator();
		while(iterator.hasNext()){
			Entry<Integer, RemoteNode> next = iterator.next();
			RemoteNode remote = next.getValue();
			if (!remote.isConnecting && !remote.isConnected()) {
				if (remote.isNeedRetry()) {
					disconnectNodes.add(remote);
				}
			}
		}
		for(RemoteNode remote : disconnectNodes){
			logger.info("begin to reconnect remoteNode:"+remote);
			remote.connect();
		}
	}
	
	public void startup() throws Exception {
		// 状态变为启动状态，停止服务的注册
		// 启动注册的服务
		List<IService> sl = new ArrayList<>();
		sl.addAll(services.values());
		Collections.sort(sl, new Comparator<IService>() {
            public int compare(IService s1, IService s2) {
                return s1.getId() - s2.getId();
            }
        });
		for (IService service : sl) {
			try {
				if (service.start()) {
					logger.info("service start ok: "+service);
				} else {
					throw new Exception("service start failed");
				}
			}catch(Exception ex){
				logger.error("服务器启动出错！service="+service+". ex="+ex, ex);
				throw new Exception("启动服务节点出错");
			}
		}
		
		// 启动监听端口
		acceptor = new NodeAcceptor(this);
		try {
			acceptor.accept();
			logger.info("服务节点绑定端口成功[nodeId={},nodeName={},port={}", id, name, port);
		} catch (Exception e) {
			logger.error("服务节点启动发送错误[nodeId={" + id + "},nodeName={" + name + "},port={" + port + "}：exception=" + e.getMessage(), e);
			throw new SysException(e, "服务节点绑定端口错误[nodeId={},nodeName={},port={}", id, name, port);
		}

		connectRemotes();
		
		logger.info("begin start auto connect timer...");
		//10秒一次检测重连
		nodeActor.regTimer(new GameTimer(10000){
			@Override
			public void doAction() {
				connectRemotes();
			}	
		});
		logger.info("start auto connect timer finish!");
	}

	/**
	 * 停止服务
	 */
	public void stopServe() {
		available = false;
		acceptor.stopAccept();
	}

	public boolean sendNodeMsg(int toNodeId, NodeMessage message) {
		try {
			//只传输WrappedMessage
			RemoteNode remote = getRemoteNode(toNodeId);
			if(remote!=null && remote.isUsable()){
				remote.send(message);
				return true;
			}else{
				//TODO:will@添加到缓存队列，避免断线丢失数据包
				logger.error("send message failed! remote node unusable: "+toNodeId+BeanUtil.getStack());
			}
		} catch (IOException e) {
			logger.error("send message failed! err=" + e.getMessage(), e);
		}
		return false;
	}
	
	/**
	 * 已经序列化好的消息
	 * @param toNodeId
	 * @param message
	 */
	public void send(int toNodeId, IMessage message) {
		try {
			if(toNodeId==this.id){
				this.getHandler().onNodeMessageReveived(message, this.getType());
				return;
			}
			RemoteNode remoteNode = this.getRemoteNode(toNodeId);
			//连接断开或者未连接上，一些重要的消息需要记录下来，比如充值的消息不能丢掉
			if(remoteNode==null){
				logger.error("未注册对应的远程节点，消息无法送达！nodeId="+toNodeId+",msgId="+message.getId()+",msg="+message);
				
			}else if(!remoteNode.isUsable()){
				if(!remoteNode.isConnected()){
					logger.error("远程节点已断开连接，消息无法送达！nodeId="+toNodeId+",msgId="+message.getId()+",msg="+message);
				}else{
					logger.error("远程节点已连接，但未注册成功，消息无法送达！nodeId="+toNodeId+",msgId="+message.getId()+",msg="+message);
				}
			}
			RemoteNode remote = getRemoteNode(toNodeId);
			if(remote!=null && remote.isUsable()){
				remote.send(message);
			}else{
				//发送失败！
			}
		} catch (IOException e) {
			logger.error("send message failed! err=" + e.getMessage(), e);
		}
	}
	
	
	public void send(int serverId, NodeType nodeType, Object message) {
		try {
			List<RemoteNode> toNodes = new ArrayList<>();
			for(RemoteNode remote : remoteNodes.values()){
				if(remote.getServerId() == serverId && 
				   remote.type==nodeType){
					toNodes.add(remote);
				}
			}
			for(RemoteNode remote : toNodes){
				remote.send(message);
			}
		} catch (IOException e) {
			logger.error("send message failed! err=" + e.getMessage(), e);
		}
	}
	
	/**
	 * 发送消息到指定的服务节点
	 * @param toNodeId
	 * @param sender
	 * @param message
	 */
	public void send(int toNodeId, long sender, IMessage message) {
		// 只传输WrappedMessage
		send(toNodeId,sender,message,null);
	}
	
	/**
	 * 发送消息到指定的服务节点
	 * @param toNodeId
	 * @param sender
	 * @param message
	 */
	public void send(int toNodeId, long sender, IMessage message, List<Long> receivers) {
		message.setSender(sender);
		if(receivers!=null && !receivers.isEmpty()){
			message.getReceivers().addAll(receivers);
		}
		send(toNodeId, message);
	}

	public Map<Integer, RemoteNode> getRemoteNodes() {
		return remoteNodes;
	}

	public List<RemoteNode> getRemoteNodes(NodeType type) {
		List<RemoteNode> r = new ArrayList<>();
		for (RemoteNode n : remoteNodes.values()) {
			if (n.getType() == type) {
				r.add(n);
			}
		}
		return r;
	}
	
	/**
	 * 添加远程子节点
	 * 
	 * @param node
	 */
	public void addRemoteNode(RemoteNode node) {
		node.setNeedRetry(needReconnect(node));
		node.setLocalNode(this);
		remoteNodes.put(node.getId(), node);
	}

	public RemoteNode getRemoteNode(int remoteId) {
		return remoteNodes.get(remoteId);
	}
	
	public void rmRemoteNode(int nodeId) {
		remoteNodes.remove(nodeId);
	}

	public String getNodeKey() {
		return host + ":" + port;
	}

	public NodeType getType() {
		return type;
	}

	/**
	 * 初始化节点配置
	 * @param configPath
	 * @throws ConfigException 
	 */
	public static void initConfig(int serverId, String configPath) throws ConfigException{
		NodeConfig.loadConfig(serverId, configPath);
	}

	/**
	 * 初始化服务节点配置
	 */
	public static Node createNode(String serverName) {
		Node node = null;
		NodeConfig config = NodeConfig.getConfig(serverName);
		if (config != null) {
			Set<String> tmp = new HashSet<>();
			for (String name : NodeConfig.allNodesName()) {
				if (!name.equalsIgnoreCase(serverName)) {
					tmp.add(name);
				}
			}
			config.setRemoteNodes(tmp);
			node = createNode(config);
		} else {
			String errInfo = "当前服务器的服务节点配置未找到！serverName=" + serverName;
			logger.error(errInfo);
		}
		return node;
	}
	
	/**
	 * 初始化服务节点配置
	 */
	public static Node createNode(NodeConfig config) {
		Node node = new Node(config);
		for (String remote : config.getRemoteNodes()) {
			NodeConfig remoteConfig = NodeConfig.getConfig(remote);
			if (remoteConfig != null) {
				RemoteNode remoteNode = new RemoteNode(remoteConfig.getHost(), remoteConfig.getPort());
				remoteNode.setId(remoteConfig.getId());
				remoteNode.setName(remoteConfig.getName());
				remoteNode.type=remoteConfig.getType();
				node.addRemoteNode(remoteNode);
			}
		}
		return node;
	}
	
	/**
	 * 重连规则：
	 * <pre>
	 * 结点ID小的会主动连接比自己大的结点
	 * </pre>
	 * @param nodeType
	 * @param remoteType
	 * @return
	 */
	public boolean needReconnect(RemoteNode remoteNode){
		if (getType() == NodeType.GAME) {
			if (remoteNode.getType() == NodeType.MATCH ||
				remoteNode.getType() == NodeType.CROSS ||
				remoteNode.getType() == NodeType.CENTER) {
				return true;
			}
		} else if (getType() == NodeType.CROSS) {
			if (remoteNode.getType() == NodeType.MATCH) {
				return true;
			}
		}
		return false;
	}

	public void addThreadContext(ThreadContext context){
		threadContexts.put(context.getId(), context);
	}
	
	public void removeThreadContext(ThreadContext context){
		threadContexts.remove(context.getId());
	}
	
	public int applyThreadContextId(){
		return threadContextIdGen.getAndAdd(1);
	}
	
	/**
	 * 注册服务
	 */
	public static void registerService(IService service){
		int serviceId = service.getId();
		if(services.containsKey(serviceId)){
			String errText = "register service failed: serviceId repeat! serviceId="+serviceId+",service="+service;
			logger.error(errText);
			throw new SysException(errText);
		}
		services.put(serviceId, service);
		//注册到服务定位器中
		ServiceLocator.register(service.getId(), service);
	}
	
	/**
	 * 移除服务注册
	 * @param service
	 */
	public static  void unregisterService(IService service){
		unregisterService(service.getId());
	}
	
	/**
	 * 根据id获取服务
	 * @param serviceId
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static <T extends IService> T getService(int serviceId){
		return (T)services.get(serviceId);
	}
	
	/**
	 * 根据服务id移除注册的服务
	 * @param serviceId
	 */
	public static void unregisterService(int serviceId){
		services.remove(serviceId);
	}

	/**
	 * 让对应的线程处理远程调用
	 * @param call
	 */
	public boolean addCall(Call call) {
		IService service = services.get(call.serviceId);
		if(service==null){
			logger.error("service not registered on this node! serviceId="+call.serviceId);
			Context.tryReturn(call, null);
			return false;
		}
		service.doCall(call);
		return true;
	}

	/**
	 * 处理远程调用返回的结果
	 * @param callReturn
	 */
	public void addCallReturn(CallReturn callReturn) {
		if (callReturn.isActor) {
			Context.callRet(callReturn);
		} else {
			ThreadContext processor = threadContexts.get(callReturn.threadKey);
			if(processor==null){
				return;
			}
			processor.onReturn(callReturn);
		}
	}
	
	public void onNodeRegister(RemoteNode n) {
		for (IService s : services.values()) {
			if (s.getNodeNotify()) {
				s.addAction(()->{
					s.onNodeRegister(n.getId(), n.getName(), n.getHost(), n.getPort(), n.getType());
				});
			}
		}
	}
	
	public void onNodeUnregister(RemoteNode n) {
		for (IService s : services.values()) {
			if (s.getNodeNotify()) {
				s.addAction(()->{
					s.onNodeUnregister(n.getId(), n.getType());
				});
			}
		}
	}
}
